package com.iwechat.adapter.mina;

import java.nio.charset.Charset;

import javax.annotation.PostConstruct;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.proxy.utils.ByteUtilities;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.iwechat.adapter.mina.entity.Active;
import com.iwechat.adapter.mina.entity.ActiveResp;
import com.iwechat.adapter.mina.entity.Connect;
import com.iwechat.adapter.mina.entity.ConnectResp;
import com.iwechat.adapter.mina.entity.Exit;
import com.iwechat.adapter.mina.entity.ExitResp;
import com.iwechat.adapter.mina.entity.PDU;
import com.iwechat.adapter.mina.entity.Submit;
import com.iwechat.adapter.mina.entity.SubmitResp;
import com.iwechat.emuns.ConstEmus;
import com.iwechat.mybatis.generator.model.TemplateMessage;
import com.iwechat.service.ClientFactory;
import com.iwechat.service.MessageHandlerFactory;

public class NioHandlerAdapter extends IoHandlerAdapter {
    private static final Logger logger         = LogManager.getLogger();
    private static final String ISLOGIN        = "_isLogin?_";
    private static final String MAX_WINDOW     = "_mwindow?_";
    private static final String CURRENT_WINDOW = "_cwindow?_";
    private String              charset;
    private Charset             charSet;

    @Autowired
    private ClientFactory       cfactory;
    @Autowired
    MessageHandlerFactory       msgHandlerFac;

    @PostConstruct
    public void init() {
        this.charSet = Charset.forName(charset);
        logger.info("### " + this + " init succeed!");
    }

    @Override
    public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
        logger.error("sessionId[" + session.getId() + "] exceptionCaught:", cause);
        session.closeNow();
    }

    @Override
    public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
        logger.info("sessionId[" + session.getId() + "] sessionIdle.");
        Active active = new Active();
        session.write(active);
    }

    @Override
    public void messageReceived(IoSession session, Object message) throws Exception {
        logger.info("sessionId[" + session.getId() + "] messageReceived:{}", message);
        if (logger.isDebugEnabled()) {
            logger.debug("sessionId[" + session.getId() + "] messageReceived:{}", ByteUtilities.asHex(((PDU) message).toIoBuffer(charSet).array()));
        }
        if (message instanceof Connect) {
            doConnect(session, (Connect) message);
        } else if (message instanceof Submit) {
            doSubmit(session, (Submit) message);
        } else if (message instanceof Active) {
            doActive(session, (Active) message);
        } else if (message instanceof Exit) {
            doExit(session, (Exit) message);
        } else if (message instanceof ExitResp) {
            session.closeNow();
        }
    }

    /**
     * 处理登录请求
     *
     *
     * @param session
     * @param connect
     * @throws Exception
     */
    private void doConnect(IoSession session, Connect connect) throws Exception {
        ConnectResp resp = new ConnectResp(connect, PDU.ERRORCODE_SUCCESS);

        String clientID = connect.getClientID().trim();
        String publicKey = cfactory.getPublicStringByClientId(clientID);

        boolean verify = false;
        try {
            verify = connect.verify(publicKey, charSet);
        } catch (Exception e) {
            logger.info(e, e);
        }
        if (!verify) {
            resp.setErrorCode(PDU.ERRORCODE_CONN_VERIFY_FAIL);
        } else {
            session.setAttribute(ISLOGIN, verify);
            // 登录后设置当前窗口大小为允许的最大窗口
            int maxWindow = cfactory.getWindowByClientId(clientID);
            session.setAttribute(MAX_WINDOW, maxWindow);
            session.setAttribute(CURRENT_WINDOW, maxWindow);
        }
        session.write(resp);
    }

    /**
     * 处理消息发送请求
     *
     *
     * @param session
     * @param message
     */
    private void doSubmit(IoSession session, Submit message) {
        SubmitResp resp = new SubmitResp(message, PDU.ERRORCODE_SUCCESS);
        // window--
        minusWindow(session, message);

        boolean isLogin = false;
        if (session.containsAttribute(ISLOGIN)) {
            isLogin = (Boolean) session.getAttribute(ISLOGIN);
        }
        if (isLogin) {
            try {
                String body = new String(message.getBody(), this.charSet);
                JSONObject data = JSON.parseObject(body);
                String clientIp = (String) session.getAttribute(ConstEmus.clientIp.value());
                String clientId = (String) session.getAttribute(ConstEmus.clientId.value());
                data.put(ConstEmus.clientIp.value(), clientIp);
                data.put(ConstEmus.clientId.value(), clientId);
                data.put(ConstEmus.sessionId.value(), session.getId());
                data.put(ConstEmus.sequence.value(), message.getSequence());
                logger.info("sessionId[" + session.getId() + "] for prepare: " + data.toJSONString());
                // 上送消息进行模板预处理
                TemplateMessage tm = msgHandlerFac.prepare(data);
                if (tm == null) {
                    // 预处理失败
                    resp.setErrorCode(PDU.ERRORCODE_DATA_PREPARE_ERROR);
                } else {
                    // 模板消息处理
                    int result = msgHandlerFac.doHandle(tm);
                    resp.setErrorCode(result);
                }

            } catch (Exception e) {
                logger.error(e, e);
            }
        } else {
            // 非法消息,未登录
            resp.setErrorCode(PDU.ERRORCODE_SUB_NO_LOGIN);
        }
        // window++
        plusWindow(session);
        // 窗口恢复后发送响应
        session.write(resp);
    }

    /**
     * 接收一条消息后,可用窗口-1
     *
     *
     * @param submit
     * @param session
     */
    private void minusWindow(IoSession session, Submit submit) {
        synchronized (session) {
            int current = 0;
            if (session.containsAttribute(CURRENT_WINDOW)) {
                current = (Integer) session.getAttribute(CURRENT_WINDOW);
            } else {
                current = 0;
            }
            // 可用窗口为0
            if (current <= 0) {
                int max = (Integer) session.getAttribute(MAX_WINDOW);
                logger.error("sessionId[" + session.getId() + "]该会话接收窗口[" + max + "]已满,客户端发送过快,进行流量控制。");
                SubmitResp resp = new SubmitResp(submit, PDU.ERRORCODE_SUB_TOO_FAST);
                session.write(resp);
            } else {
                current--;
                session.setAttribute(CURRENT_WINDOW, current);
            }
        }
    }

    /**
     * 消息处理完成后,可用窗口+1
     *
     *
     * @param session
     */
    private void plusWindow(IoSession session) {
        synchronized (session) {
            int current = 0;
            if (session.containsAttribute(CURRENT_WINDOW)) {
                current = (Integer) session.getAttribute(CURRENT_WINDOW);
            } else {
                current = 0;
            }
            int max = (Integer) session.getAttribute(MAX_WINDOW);
            if (current < max) {
                current++;
            } else {
                current = max;
            }
            session.setAttribute(CURRENT_WINDOW, current);
        }
    }

    /**
     * 处理登出请求
     *
     *
     * @param session
     * @param message
     */
    private void doExit(IoSession session, Exit message) {
        ExitResp resp = new ExitResp(message);
        session.write(resp);

    }

    /**
     * 处理心跳请求
     *
     *
     * @param session
     * @param message
     */
    private void doActive(IoSession session, Active message) {
        ActiveResp resp = new ActiveResp(message);
        session.write(resp);

    }

    @Override
    public void messageSent(IoSession session, Object message) throws Exception {
        logger.info("sessionId[" + session.getId() + "] messageSent:{}", message);
    }

    @Override
    public void sessionClosed(IoSession session) throws Exception {
        logger.info("sessionId[" + session.getId() + "] sessionClosed.");
    }

    @Override
    public void sessionCreated(IoSession session) throws Exception {
        logger.info("sessionId[" + session.getId() + "] sessionCreated.");
    }

    @Override
    public void sessionOpened(IoSession session) throws Exception {
        logger.info("sessionId[" + session.getId() + "] sessionOpened.");
    }

    public void setCharset(String charset) {
        this.charset = charset;
    }
}
